package org.senRPC.client.consumer;

import static org.senRPC.common.serialization.SerializerHolder.serializerImpl;
import io.netty.channel.Channel;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.senRPC.client.consumer.NotifyListener.NotifyEvent;
import org.senRPC.common.exception.remoting.RemotingSendRequestException;
import org.senRPC.common.exception.remoting.RemotingTimeoutException;
import org.senRPC.common.loadbalance.LoadBalanceStrategy;
import org.senRPC.common.protocal.senRPCProtocol;
import org.senRPC.common.rpc.RegisterMeta;
import org.senRPC.common.transport.body.AckCustomBody;
import org.senRPC.common.transport.body.SubcribeResultCustomBody;
import org.senRPC.common.transport.body.SubscribeRequestCustomBody;
import org.senRPC.remoting.model.RemotingTransporter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *
 * @author senrian
 * @description 消费者端的一些逻辑处理,主要还是订阅服务的功能多点
 * @time 2020年2月22日
 * @modifytime
 */
public class ConsumerManager {
	//以下服务都是
	private static final Logger logger = LoggerFactory.getLogger(ConsumerManager.class);
	private DefaultConsumer defaultConsumer; //ConsumerManager的代码手持defaultConsumer好办事
	private final ReentrantReadWriteLock registriesLock = new ReentrantReadWriteLock();
	//为防止并发问题使用concurenthashmap
	private final Map<String, List<RegisterMeta>> registries = new ConcurrentHashMap<String, List<RegisterMeta>>();
	//服务名和listerner的map
	private ConcurrentHashMap<String, NotifyListener> serviceMatchedNotifyListener = new ConcurrentHashMap<String, NotifyListener>();
	private long timeout;

	public ConsumerManager(DefaultConsumer defaultConsumer) {
		this.defaultConsumer = defaultConsumer;
		this.timeout = this.defaultConsumer.getConsumerConfig().getRegistryTimeout();
	}
	//订阅服务:
	//
	public void subcribeService(String serviceName, NotifyListener listener) {

		if(listener != null){
			serviceMatchedNotifyListener.put(serviceName, listener);
		}
		if (this.defaultConsumer.getRegistyChannel() == null) {
			this.defaultConsumer.getOrUpdateHealthyChannel();
		}
		if (this.defaultConsumer.getRegistyChannel() != null) {

			logger.info("registry center channel is [{}]", this.defaultConsumer.getRegistyChannel());

			SubscribeRequestCustomBody body = new SubscribeRequestCustomBody();
			body.setServiceName(serviceName);
			//传输体只有body就行,不用传输其他的端口地址等控制数据
			RemotingTransporter remotingTransporter = RemotingTransporter.createRequestTransporter(senRPCProtocol.SUBSCRIBE_SERVICE, body);
			try {
				// 发送订阅的服务的名字, 收到返回的response是注册中新的回应,是字节
				RemotingTransporter response = sendKernelSubscribeInfo(this.defaultConsumer.getRegistyChannel(), remotingTransporter, timeout);

				// 回复注册中心ack,并将上一步的response解析出registermeta放到全局map中
				RemotingTransporter ackTransporter = this.defaultConsumer.getConsumerManager().handlerSubcribeResult(response,
						this.defaultConsumer.getRegistyChannel());
				this.defaultConsumer.getRegistyChannel().writeAndFlush(ackTransporter);
			} catch (Exception e) {
				logger.warn("registry failed [{}]", e.getMessage());
			}

		} else {
			logger.warn("sorry can not connection to registry address [{}],please check your registry address", this.defaultConsumer.getRegistryClientConfig()
					.getDefaultAddress());
		}

	}

	private RemotingTransporter sendKernelSubscribeInfo(Channel registyChannel, RemotingTransporter remotingTransporter, long timeout)
			throws RemotingTimeoutException, RemotingSendRequestException, InterruptedException {
		return this.defaultConsumer.getRegistryNettyRemotingClient().invokeSyncImpl(this.defaultConsumer.getRegistyChannel(), remotingTransporter, timeout);
	}

	public ConcurrentHashMap<String, NotifyListener> getServiceMatchedNotifyListener() {
		return serviceMatchedNotifyListener;
	}

	public void setServiceMatchedNotifyListener(ConcurrentHashMap<String, NotifyListener> serviceMatchedNotifyListener) {
		this.serviceMatchedNotifyListener = serviceMatchedNotifyListener;
	}




	/**
	 * 处理服务的订阅结果,主要有两个工作要做:封装返回ack,通知服务端将
	 * @param request
	 * @param channel
	 * @return
	 */
	public RemotingTransporter handlerSubcribeResult(RemotingTransporter request, Channel channel) {

		if (logger.isDebugEnabled()) {
			logger.debug("handler subcribe result [{}] and channel [{}]", request, channel);
		}
		AckCustomBody ackCustomBody = new AckCustomBody(request.getOpaque(), false);
		RemotingTransporter responseTransporter = RemotingTransporter.createResponseTransporter(senRPCProtocol.ACK, ackCustomBody, request.getOpaque());
		SubcribeResultCustomBody subcribeResultCustomBody = serializerImpl().readObject(request.bytes(), SubcribeResultCustomBody.class);

		String serviceName = null;
		if (subcribeResultCustomBody != null && subcribeResultCustomBody.getRegisterMeta() != null && !subcribeResultCustomBody.getRegisterMeta().isEmpty()) {
			for (RegisterMeta registerMeta : subcribeResultCustomBody.getRegisterMeta()) {
				if (null == serviceName) {
					serviceName = registerMeta.getServiceName();
				}
				notify(serviceName, registerMeta, NotifyEvent.CHILD_ADDED);
			}
		}

		ackCustomBody.setSuccess(true);
		return responseTransporter;
	}

	/**
	 * 处理服务取消的时候逻辑处理
	 *
	 * @param request
	 * @param channel
	 * @return
	 */
	public RemotingTransporter handlerSubscribeResultCancel(RemotingTransporter request, Channel channel) {
		AckCustomBody ackCustomBody = new AckCustomBody(request.getOpaque(), false);
		RemotingTransporter responseTransporter = RemotingTransporter.createResponseTransporter(senRPCProtocol.ACK, ackCustomBody, request.getOpaque());

		SubcribeResultCustomBody subcribeResultCustomBody = serializerImpl().readObject(request.bytes(), SubcribeResultCustomBody.class);

		if (subcribeResultCustomBody != null && subcribeResultCustomBody.getRegisterMeta() != null && !subcribeResultCustomBody.getRegisterMeta().isEmpty()) {

			for (RegisterMeta registerMeta : subcribeResultCustomBody.getRegisterMeta()) {
				notify(registerMeta.getServiceName(), registerMeta, NotifyEvent.CHILD_REMOVED);
			}
		}
		ackCustomBody.setSuccess(true);
		return responseTransporter;
	}

	/**
	 * 处理注册中心发送过来的负载均衡策略的变化
	 * @param request
	 * @param channel
	 * @return
	 */
	public RemotingTransporter handlerServiceLoadBalance(RemotingTransporter request, Channel channel) {

		if(logger.isDebugEnabled()){
			logger.debug("handler change loadBalance strategy [{}] and channel [{}]",request,channel);
		}
		AckCustomBody ackCustomBody = new AckCustomBody(request.getOpaque(), false);
		RemotingTransporter responseTransporter = RemotingTransporter.createResponseTransporter(senRPCProtocol.ACK, ackCustomBody, request.getOpaque());
		SubcribeResultCustomBody subcribeResultCustomBody = serializerImpl().readObject(request.bytes(), SubcribeResultCustomBody.class);
		String serviceName = subcribeResultCustomBody.getServiceName();
		LoadBalanceStrategy balanceStrategy = subcribeResultCustomBody.getLoadBalanceStrategy();
		defaultConsumer.setServiceLoadBalanceStrategy(serviceName, balanceStrategy);
		ackCustomBody.setSuccess(true);
		return responseTransporter;
	}

	// 实际上是调用listener的notify()
	private void notify(String serviceName, RegisterMeta registerMeta, NotifyEvent event) {
		boolean notifyNeeded = false;
		final Lock writeLock = registriesLock.writeLock();
		writeLock.lock();
		try {
			//一个
			List<RegisterMeta> registerMetas = registries.get(serviceName);
			if (registerMetas == null) {
				if (event == NotifyEvent.CHILD_REMOVED) {
					return;
				}
				registerMetas = new ArrayList<RegisterMeta>();
				//registermeta就是服务注解中标注的
				registerMetas.add(registerMeta);
				notifyNeeded = true;
			} else {
				if (event == NotifyEvent.CHILD_REMOVED) {
					registerMetas.remove(registerMeta);
				} else if (event == NotifyEvent.CHILD_ADDED) {
					registerMetas.add(registerMeta);
				}
				notifyNeeded = true;
			}
			registries.put(serviceName, registerMetas);
		} finally {
			writeLock.unlock();
		}

		if (notifyNeeded) {
			NotifyListener listener = this.getServiceMatchedNotifyListener().get(serviceName);
			if (null != listener) {
				listener.notify(registerMeta, event);
			}
		}
	}
}
